package com.des1gn.retry;

import com.alibaba.fastjson.JSONObject;
import com.des1gn.common.bean.RetryErrorMqMessageDto;
import com.des1gn.common.constant.MqConstants;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.lang.reflect.Method;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * @ClassName RetryProcessorManage
 * @Description 业务故障本地自动重试处理器
 * @Author Des1gn
 * @Version 1.0
 **/
@Component
@Slf4j
@RequiredArgsConstructor
public class RetryProcessorManage implements InitializingBean {

    private final ApplicationContext applicationContext;

    private final KafkaTemplate kafkaTemplate;

    public static ThreadLocal<String> currentMehodThreadLocal = new ThreadLocal<>();

    @Getter
    private final static BlockingQueue<RetryCounter> queue = new LinkedBlockingQueue<>();

    public static void push(RetryCounter retryCounter){
        queue.add(retryCounter);
    }

    public static String getCurrentMehodString(){
        return currentMehodThreadLocal.get();
    }

    protected static RetryCounter take(){
        try {
            return queue.take();
        } catch (InterruptedException e) {
            log.error("retry queue take error -> {}",e);
        }
        return null;
    }

    protected RetryErrorMqMessageDto generatorMessage(RetryCounter task){
        RetryClass retryClass = (RetryClass)task.getTargetClass().getAnnotation(RetryClass.class);
        if (retryClass == null){
            return null;
        }
        return RetryErrorMqMessageDto.builder()
                .targetClassName(task.getTargetClass().getInterfaces()[0].getName())
                .targetMethodName(task.getTargetMethod())
                .targetParamterClass(task.getParamterClassArges())
                .targetVersion(retryClass.version())
                .targetArgs(task.getTargetArgs())
                .build();
    }

    protected void sendMqMessage(RetryCounter task){
        RetryErrorMqMessageDto retryErrorMqMessageDto = generatorMessage(task);
        if (retryErrorMqMessageDto == null){
            return;
        }
        kafkaTemplate.send(MqConstants.RETRY_ERROR_GROUP,JSONObject.toJSONString(retryErrorMqMessageDto));
    }

    protected String getMethodKey(Method method,RetryCounter task){
        //设置当前处理方法
        StringBuilder methodKey = new StringBuilder();
        methodKey.append(method.toString());
        methodKey.append("->");
        methodKey.append(JSONObject.toJSONString(task.getTargetArgs()));
        return methodKey.toString();
    }

    @Override
    public void afterPropertiesSet(){
        new Thread(()->{
            while (true){
                RetryCounter task = RetryProcessorManage.take();
                if (null == task){
                    continue;
                }
                if (task.getCounter() == task.getMaxCount().intValue()){
                    //发送mq定义为需手动处理故障
                    sendMqMessage(task);
                    continue;
                }
                task.setCounter(task.getCounter()+1);
                log.info("重试第{}次",task.getCounter());
                try {
                    //获取目标调用方法
                    Object bean = applicationContext.getBean(task.getTargetClass());
                    Method method = task.getTargetClass().getMethod(task.getTargetMethod(), task.getParamterClassArges());
                    //方法调用重试
                    method.invoke(bean,task.getTargetArgs());
                }catch (Exception e){
                    log.error("retry apply exception -> {}",e);
                    RetryProcessorManage.push(task);
                }finally {
                    currentMehodThreadLocal.remove();
                }
            }
        }).start();
    }
}
